package windows;

import beans.SenSorReading;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

public class EventTimeWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置流式时间的语义
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置时间间隔
        env.getConfig().setAutoWatermarkInterval(100);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SenSorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SenSorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        })
                /**
                 * 传输出时间戳即可
                 * 给的参数是延迟时间
                 */
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SenSorReading>(Time.seconds(3)) {
                    @Override
                    public long extractTimestamp(SenSorReading senSorReading) {
                        return senSorReading.getTimeStamp() * 1000l;
                    }
                });
        /**
         * 这个是默认的，不需要给延迟时间，也就是说如果数据时有序的话，就用这个
         */
//                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SenSorReading>() {
//                    @Override
//                    public long extractAscendingTimestamp(SenSorReading senSorReading) {
//                        return senSorReading.getTimeStamp() * 1000l;
//                    }
//                });
        //基于事件时间的开窗聚合
        //统计15s内温度的最小值
        SingleOutputStreamOperator<SenSorReading> minTempture = dataStream.keyBy("id")
                .timeWindow(Time.seconds(15))
                .minBy("temperature");

        minTempture.print();
        env.execute();
    }
}
